package com.tianji.mq.consumer;

import com.tianji.common.constant.MqConstants;
import com.tianji.common.handler.MessageQueueIdempotentHandler;
import com.tianji.common.lock.MyLock;
import com.tianji.common.lock.MyLockStrategy;
import com.tianji.common.lock.MyLockType;
import com.tianji.coupon.mapper.CouponMapper;
import com.tianji.coupon.pojo.Coupon;
import com.tianji.coupon.service.UserCouponService;
import com.tianji.exception.BizException;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Map;

import static com.tianji.common.constant.MqConstants.CONSUMER;

@Service
@RocketMQMessageListener(topic = MqConstants.EXCHANGE,
        selectorExpression = MqConstants.ROUTING_KEY,
        consumerGroup = CONSUMER)
@RequiredArgsConstructor
public class CounponConsumer implements RocketMQListener<Map<String,String>> {

    private final CouponMapper couponMapper;

    private final UserCouponService userCouponService;

    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @MyLock(
            name = "lock:consumer:#{message['userId']}",  // 使用 SpEL 获取 userId
            lockType = MyLockType.RE_ENTRANT_LOCK,
            lockStrategy = MyLockStrategy.FAIL_AFTER_RETRY_TIMEOUT
    )
    @Override
    @Transactional
    public void onMessage(Map<String, String> message) {
        String userId = message.get("userId");
        String couponId = message.get("couponId");
        try{
            //消息幂等性处理，检查是否存在消息,返回true证明消息已经存在了
            if(messageQueueIdempotentHandler.isMessageBeingConsumed(couponId)){
                //判断当前消息是否已经处理完成,true证明已经消费完成
                if(messageQueueIdempotentHandler.isAccomplish(couponId)){
                    return;
                }
                throw new RuntimeException("消息未完成，需要消息队列重试");
            }
            //更新优惠券已经发放的数量
            Coupon coupon = couponMapper.selectById(couponId);
            int i = couponMapper.incrIssueNum(Long.parseLong(couponId),coupon.getIssueNum(),coupon.getUserLimit());
            if(i == 0){
                throw new BizException("优惠券库存不足");
            }
            //新增一个用户券
            userCouponService.saveUser(coupon,Long.parseLong(userId));
        }catch (Exception e){
            //删除已经消费的消息，避免重复消费
            messageQueueIdempotentHandler.delMessageProcessed(couponId);
            throw new RuntimeException("新增用户优惠券信息失败"+e);
        }
        //标记消息已经完成
        messageQueueIdempotentHandler.setAccomplish(couponId);
    }
}
